<?php
declare(strict_types=1);

namespace app\process;

use Workerman\Timer;                                  // Workerman 定时器
use Workerman\Worker;                                 // Workerman Worker 对象
use support\Redis;                                    // Webman Redis 封装
use Illuminate\Database\Capsule\Manager as DB;        // Eloquent/DB 批量插入
use app\service\CoordinateService;                    // 坐标转换服务

/**
 * GPS 数据入库消费进程（Eloquent/DB 批量写入）
 *
 * 职责：
 * - 从 Redis 列表 gps:json 批量 rpop 获取 NDJSON 行（生产者在 GpsTcp lpush）
 * - 按 config/gps_storage.php 的字段映射转换为目标表行
 * - 若目标表不存在，按映射自动建表（JSON 不支持时降级为 TEXT）
 * - 定时批量写入 MySQL（Eloquent/DB::table()->insert）
 *
 * 关键参数（.env）：
 * - GPS_SAVE_ENABLE=1            是否启用入库进程（1 开启 / 0 关闭）
 * - GPS_SAVE_WORKERS=1           进程数（根据写入压力调整）
 * - GPS_SAVE_BATCH=200           每批最大写入条数
 * - GPS_SAVE_INTERVAL_MS=200     拉取与写库周期（毫秒）
 * - GPS_TABLE=gps_positions      目标表名（可在 config/gps_storage.php 指定）
 */
class GpsSaver
{
    /** @var Worker 当前 Worker（可用于扩展） */
    protected Worker $worker;

    /** @var array 存储配置（table/columns/indexes） */
    protected array $storage = [];
    protected CoordinateService $coord;                 // 公共坐标转换服务

    /**
     * Worker 启动入口
     * - 读取存储配置
     * - 自动建表（如不存在）
     * - 定时批量消费 Redis→DB
     */
    public function onWorkerStart(Worker $worker): void
    {
        $this->worker = $worker;
        $this->coord = new CoordinateService();        // 初始化坐标服务

        // 1) 读取存储配置（字段映射/索引等）
        $cfg = config('gps_storage', []);
        $this->storage = is_array($cfg) ? $cfg : [];
        // 基础表名（不带前缀，供 DB::table 使用，让连接配置自动加前缀）
        $baseTable = $this->storage['table'] ?? env('GPS_TABLE', 'gps_positions');
        $prefix = (string)env('DB_PREFIX', '');
        if ($prefix !== '' && stripos($baseTable, $prefix) === 0) {
            // 若用户在配置里写了带前缀的表名，去掉一次，避免 DB 层再次自动加前缀导致重复
            $baseTable = substr($baseTable, strlen($prefix));
        }
        // 当月路由表名（base_YYYYMM），DDL 用真实表名（手动拼接一次前缀）
        $currYm = date('Ym');
        $routeBase = $baseTable . '_' . $currYm;
        $ddlTable = $prefix !== '' ? ($prefix . $routeBase) : $routeBase;

        // 2) 自动建当月表（若不存在）
        $this->ensureTable($ddlTable);
        echo "[Saver] Partition=month Using base={$baseTable}, current_table={$routeBase} (auto route by reported_at)\n";

        // 3) 周期性批量消费（最小 10ms，避免过于频繁）
        $interval = max(10, (int)env('GPS_SAVE_INTERVAL_MS', 200));
        Timer::add($interval / 1000.0, function () use ($baseTable) {
            $this->consumeOnce($baseTable); // 消费时基于 reported_at 动态路由到 base_YYYYMM
        });
    }

    /**
     * 自动创建数据表（若不存在）
     * - 使用 DDL 语句（DB::statement），适配非 Laravel 场景
     * - JSON 不支持时，将 JSON 定义替换为 TEXT 再建表
     */
    protected function ensureTable(string $ddlTable): void
    {
        // 简易存在性检查：尝试执行 SELECT，若抛异常则认为不存在
        try {
            DB::table($ddlTable)->limit(1)->get();
            return; // 表已存在
        } catch (\Throwable $e) {
            // 继续创建
        }

        $cols = $this->storage['columns'] ?? [];
        if (!$cols) return;

        // 组装列定义
        $defs = [];
        foreach ($cols as $c) {
            $name = $c['name'] ?? '';
            $type = $c['type'] ?? 'VARCHAR(255) NULL';
            if (!$name) continue;
            // 字段备注（可选）：config/gps_storage.php 的 columns 内可提供 'comment'
            $comment = isset($c['comment']) ? (string)$c['comment'] : '';
            $commentSql = $comment !== '' ? " COMMENT '" . str_replace("'", "\\'", $comment) . "'" : '';
            $defs[] = sprintf("`%s` %s%s", $name, $type, $commentSql);
        }

        // 拼接 DDL（包含主键与索引）
        $ddl = "CREATE TABLE IF NOT EXISTS `{$ddlTable}` (\n"
             . "  `id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,\n"
             . "  " . implode(",\n  ", $defs) . ",\n"
             . "  PRIMARY KEY (`id`)"
             ;

        $indexes = $this->storage['indexes'] ?? [];
        foreach ($indexes as $idx) {
            $ddl .= ",\n  {$idx}";
        }
        // 表备注（可选）：config/gps_storage.php 可提供 'table_comment'
        $tableComment = isset($this->storage['table_comment']) ? (string)$this->storage['table_comment'] : '';
        $tableCommentSql = $tableComment !== '' ? " COMMENT='" . str_replace("'", "\\'", $tableComment) . "'" : '';
        $ddl .= "\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4{$tableCommentSql};";

        // 执行建表，必要时降级 JSON→TEXT
        try {
            echo "[Saver] CREATE TABLE IF NOT EXISTS `{$ddlTable}` ...\n";
            DB::statement($ddl);
            echo "[Saver] CREATE TABLE ok for {$ddlTable}\n";
        } catch (\Throwable $e) {
            $msg = $e->getMessage();
            echo "[Saver] CREATE TABLE error: {$msg}\n";
            // 若 JSON 类型不支持，降级为 TEXT 再试一次
            $ddl2 = str_ireplace(' JSON ', ' TEXT ', $ddl);
            try {
                echo "[Saver] RETRY CREATE with TEXT for {$ddlTable}\n";
                DB::statement($ddl2);
                echo "[Saver] CREATE TABLE ok (TEXT fallback) for {$ddlTable}\n";
            } catch (\Throwable $e2) {
                echo "[Saver] CREATE TABLE retry error: {$e2->getMessage()}\n";
                Redis::set('gps:last_error', 'ddl_error:' . substr($e2->getMessage(), 0, 200));
            }
        }
    }

    /**
     * 消费一次：从 Redis 批量取数据并写库
     * - 生产端使用 lpush，为保持队列先入先出，这里使用 rpop
     */
    protected function consumeOnce(string $baseTable): void
    {
        // 批量大小（至少 1 条）
        $batch = max(1, (int)env('GPS_SAVE_BATCH', 200));

        // 从 Redis 批量取行（字符串格式 JSON）
        $rawLines = $this->drainRedis('gps:json', $batch);
        if (!$rawLines) {
            return; // 无数据则返回
        }

        // 读取列映射配置
        $columns = $this->storage['columns'] ?? [];
        if (!$columns) {
            return;
        }

        // 将 JSON 文本映射为待入库的行数组
        $rows = $this->mapRows($rawLines, $columns);
        if (!$rows) {
            return;
        }

        // 按 reported_at 的月份分组，路由到 base_YYYYMM 表
        $groups = [];
        foreach ($rows as $r) {
            $rep = $r['reported_at'] ?? null;
            $ym = $rep ? date('Ym', strtotime((string)$rep)) : date('Ym');
            $groups[$ym][] = $r;
        }

        $prefix = (string)env('DB_PREFIX', '');
        $saved = 0;
        foreach ($groups as $ym => $partRows) {
            $routeBase = $baseTable . '_' . $ym;              // DB::table 用基础名，连接自动加前缀
            $ddlTable  = $prefix !== '' ? ($prefix . $routeBase) : $routeBase; // DDL 用完整表名

            // 确保当月（该分组）表存在
            $this->ensureTable($ddlTable);

            // 批量写库
            try {
                DB::table($routeBase)->insert($partRows);
                $saved += count($partRows);
            } catch (\Throwable $e) {
                $msg = $e->getMessage();
                echo "DB batch insert error({$routeBase}): {$msg}\n";
                if (stripos($msg, 'Base table or view not found') !== false || stripos($msg, '42S02') !== false) {
                    $this->ensureTable($ddlTable);
                    try {
                        DB::table($routeBase)->insert($partRows);
                        $saved += count($partRows);
                    } catch (\Throwable $e2) {
                        echo "DB retry insert error({$routeBase}): {$e2->getMessage()}\n";
                        Redis::set('gps:last_error', 'db_error_retry:' . substr($e2->getMessage(), 0, 200));
                    }
                } else {
                    Redis::set('gps:last_error', 'db_error:' . substr($msg, 0, 200));
                }
            }
        }

        if ($saved > 0) {
            Redis::incrby('gps:saved_rows', $saved);
            Redis::set('gps:last_saved_at', date('Y-m-d H:i:s'));
        }
    }

    /**
     * 从 Redis 列表批量取出（rpop，保持 FIFO）
     * @param string $key 列表键名
     * @param int $max 最大取出条数
     * @return array 返回 JSON 字符串数组
     */
    protected function drainRedis(string $key, int $max): array
    {
        $res = [];
        for ($i = 0; $i < $max; $i++) {
            $v = Redis::rpop($key);                                       // rpop 取最右侧（生产者用 lpush）
            if ($v === null) break;                                       // 无数据则停止
            $res[] = (string)$v;                                          // 收集为字符串
        }
        return $res;
    }

    /**
     * 将 JSON 文本映射为待入库的行数组
     * - 按 columns 映射 source→name；raw/created_at 特殊处理
     * - reported_at（源 time）转换为 DATETIME
     * @param array $jsonLines JSON 字符串数组
     * @param array $columns 字段映射定义
     * @return array 待入库的行数组（关联数组）
     */
    protected function mapRows(array $jsonLines, array $columns): array
    {
        $out = [];
        $now = date('Y-m-d H:i:s');

        foreach ($jsonLines as $l) {
            $src = json_decode($l, true);                                 // 解码为数组
            if (!is_array($src)) {
                continue;                                                 // 非法 JSON 跳过
            }

            // 原始 WGS84（厂家文档：lng/lat）
            $lngWgs = isset($src['lng']) ? (float)$src['lng'] : null;
            $latWgs = isset($src['lat']) ? (float)$src['lat'] : null;

            // 派生坐标（使用公共服务封装，保持 [lng, lat] 返回顺序）
            if ($lngWgs !== null && $latWgs !== null) {
                $gcjPair = $this->coord->convert($latWgs, $lngWgs, 'wgs84', 'gcj02');
                $gcj = [$gcjPair['lon'] ?? null, $gcjPair['lat'] ?? null];
                if ($gcj[0] !== null && $gcj[1] !== null) {
                    $bdPair = $this->coord->convert($gcj[1], $gcj[0], 'gcj02', 'bd09');
                    $bd = [$bdPair['lon'] ?? null, $bdPair['lat'] ?? null];
                } else {
                    $bd = [null, null];
                }
            } else {
                $gcj = [null, null];
                $bd = [null, null];
            }

            $row = [];
            foreach ($columns as $c) {
                $name = $c['name'] ?? null;                               // 目标列名
                if (!$name) continue;

                $source  = $c['source'] ?? null;                          // 源字段名
                $convert = $c['convert'] ?? null;                         // 可选转换：gcj02 | bd09

                if ($name === 'created_at') {
                    $row[$name] = $now;                                   // 统一填入当前时间
                    continue;
                }
                if ($name === 'raw') {
                    $row[$name] = json_encode($src, JSON_UNESCAPED_UNICODE); // 存原始 JSON
                    continue;
                }
                if ($source === 'time') {
                    $row[$name] = self::parseTime($src['time'] ?? null);  // 将 time 转换为 DATETIME
                    continue;
                }

                // 坐标转换支持：当配置了 convert 时，优先使用派生坐标
                if ($convert === 'gcj02') {
                    if ($name === 'longitude_gcj02') $row[$name] = $gcj[0];
                    elseif ($name === 'latitude_gcj02') $row[$name] = $gcj[1];
                    else $row[$name] = null;
                    continue;
                }
                if ($convert === 'bd09') {
                    if ($name === 'longitude_bd09') $row[$name] = $bd[0];
                    elseif ($name === 'latitude_bd09') $row[$name] = $bd[1];
                    else $row[$name] = null;
                    continue;
                }

                // 默认映射
                if ($source) {
                    // WGS84直存列支持（如 longitude_wgs84/latitude_wgs84）
                    if ($source === 'lng') $row[$name] = $lngWgs;
                    elseif ($source === 'lat') $row[$name] = $latWgs;
                    else $row[$name] = $src[$source] ?? null;
                } else {
                    $row[$name] = null;                                   // 未指定源时填 null
                }
            }

            if ($row) {
                $out[] = $row;                                            // 收集有效行
            }
        }

        return $out;
    }

    /**
     * WGS84 → GCJ-02（高德/腾讯），仅中国境内转换
     * @return array [lng, lat]
     */
    protected static function wgs84ToGcj02(float $lng, float $lat): array
    {
        if (!self::inChina($lng, $lat)) return [$lng, $lat];
        $pi = 3.14159265358979324;
        $a = 6378245.0;
        $ee = 0.00669342162296594323;

        $dLat = self::transformLat($lng - 105.0, $lat - 35.0);
        $dLng = self::transformLng($lng - 105.0, $lat - 35.0);
        $radLat = $lat / 180.0 * $pi;
        $magic = 1 - $ee * sin($radLat) * sin($radLat);
        $sqrtMagic = sqrt($magic);
        $dLat = ($dLat * 180.0) / (($a * (1 - $ee)) / ($magic * $sqrtMagic) * $pi);
        $dLng = ($dLng * 180.0) / ($a / $sqrtMagic * cos($radLat) * $pi);

        return [$lng + $dLng, $lat + $dLat];
    }

    /**
     * GCJ-02 → BD-09（百度）
     * @return array [lng, lat]
     */
    protected static function gcj02ToBd09(float $lng, float $lat): array
    {
        $pi = 3.14159265358979324;
        $x = $lng; $y = $lat;
        $z = sqrt($x * $x + $y * $y) + 0.00002 * sin($y * $pi);
        $theta = atan2($y, $x) + 0.000003 * cos($x * $pi);
        $bdLng = $z * cos($theta) + 0.0065;
        $bdLat = $z * sin($theta) + 0.006;
        return [$bdLng, $bdLat];
    }

    /** 是否在中国境内（粗略边界），境外不做加偏 */
    protected static function inChina(float $lng, float $lat): bool
    {
        return $lng > 73.66 && $lng < 135.05 && $lat > 3.86 && $lat < 53.55;
    }

    /** GCJ 加偏辅助 */
    protected static function transformLat(float $x, float $y): float
    {
        $pi = 3.14159265358979324;
        $ret = -100.0 + 2.0 * $x + 3.0 * $y + 0.2 * $y * $y + 0.1 * $x * $y + 0.2 * sqrt(abs($x));
        $ret += (20.0 * sin(6.0 * $x * $pi) + 20.0 * sin(2.0 * $x * $pi)) * 2.0 / 3.0;
        $ret += (20.0 * sin($y * $pi) + 40.0 * sin($y / 3.0 * $pi)) * 2.0 / 3.0;
        $ret += (160.0 * sin($y / 12.0 * $pi) + 320.0 * sin($y * $pi / 30.0)) * 2.0 / 3.0;
        return $ret;
    }
    protected static function transformLng(float $x, float $y): float
    {
        $pi = 3.14159265358979324;
        $ret = 300.0 + $x + 2.0 * $y + 0.1 * $x * $x + 0.1 * $x * $y + 0.1 * sqrt(abs($x));
        $ret += (20.0 * sin(6.0 * $x * $pi) + 20.0 * sin(2.0 * $x * $pi)) * 2.0 / 3.0;
        $ret += (20.0 * sin($x * $pi) + 40.0 * sin($x / 3.0 * $pi)) * 2.0 / 3.0;
        $ret += (150.0 * sin($x / 12.0 * $pi) + 300.0 * sin($x / 30.0 * $pi)) * 2.0 / 3.0;
        return $ret;
    }

    /**
     * 将厂家示例时间字符串转为 MySQL DATETIME
     * 示例："2025-08-27 18:06:52"
     * @param ?string $s 时间字符串
     * @return ?string DATETIME 格式或 null
     */
    protected static function parseTime(?string $s): ?string
    {
        if (!$s) return null;
        $t = strtotime($s);                                               // 解析为时间戳
        return $t ? date('Y-m-d H:i:s', $t) : null;                       // 转回 DATETIME
    }
}